package com.dys.flink.wc.netty;

import com.dys.flink.wc.model.Product;
import com.dys.flink.wc.util.JsonUtils;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CopyOnWriteArraySet;

public class ProductServer {
    private static final Logger logger = LoggerFactory.getLogger(ProductServer.class);
    private final int clientPort;  // 客户端连接端口
    private final int flinkPort;   // Flink 连接端口
    private final CopyOnWriteArraySet<ChannelHandlerContext> flinkClients = new CopyOnWriteArraySet<>();

    public ProductServer(int clientPort, int flinkPort) {
        this.clientPort = clientPort;
        this.flinkPort = flinkPort;
    }

    public void run() throws Exception {
        // 启动客户端服务
        startClientServer();
        // 启动 Flink 服务
        startFlinkServer();
    }

    private void startClientServer() {
        // 使用新线程启动客户端服务，避免阻塞主线程
        new Thread(() -> {
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();

            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) {
                                ChannelPipeline pipeline = ch.pipeline();
                                // 添加解码器解决粘包问题
                                pipeline.addLast(new com.dys.flink.wc.netty.handler.ProductDecoder());
                                pipeline.addLast(new com.dys.flink.wc.netty.handler.ProductEncoder());
                                pipeline.addLast(new SimpleChannelInboundHandler<Product>() {
                                    @Override
                                    protected void channelRead0(ChannelHandlerContext ctx, Product product) {
                                        logger.info("Received product from client: {}", product);

                                        // 将商品数据广播给所有连接的 Flink 客户端
                                        String jsonData = JsonUtils.toJson(product);
                                        broadcastToFlinkClients(jsonData);

                                        logger.info("Broadcast product to {} flink clients: {}", flinkClients.size(), jsonData);
                                    }

                                    @Override
                                    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                                        logger.error("Client server error", cause);
                                        ctx.close();
                                    }
                                });
                            }
                        })
                        .option(ChannelOption.SO_BACKLOG, 128)
                        .childOption(ChannelOption.SO_KEEPALIVE, true);

                ChannelFuture f = b.bind(clientPort).sync();
                logger.info("Client Server started on port {}", clientPort);
                f.channel().closeFuture().sync();
            } catch (Exception e) {
                logger.error("Client server error", e);
            } finally {
                workerGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            }
        }).start();
    }

    private void startFlinkServer() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            // 添加字符串编码器，方便发送文本数据给 Flink
                            pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
                            pipeline.addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                    flinkClients.add(ctx);
                                    logger.info("Flink client connected from {}", ctx.channel().remoteAddress());
                                    logger.info("Total flink clients: {}", flinkClients.size());
                                }

                                @Override
                                public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                                    flinkClients.remove(ctx);
                                    logger.info("Flink client disconnected");
                                    logger.info("Total flink clients: {}", flinkClients.size());
                                }

                                @Override
                                public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                                    logger.error("Flink server error", cause);
                                    flinkClients.remove(ctx);
                                    ctx.close();
                                }
                            });
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            ChannelFuture f = b.bind(flinkPort).sync();
            logger.info("Flink Server started on port {}", flinkPort);
            logger.info("Flink can connect to localhost:{} to receive product data", flinkPort);

            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

    private void broadcastToFlinkClients(String jsonData) {
        if (flinkClients.isEmpty()) {
            logger.warn("No flink clients connected to receive data");
            return;
        }

        // 添加换行符，因为 Flink 的 socketTextStream 按行读取
        String dataWithNewline = jsonData + "\n";

        int sentCount = 0;
        for (ChannelHandlerContext ctx : flinkClients) {
            if (ctx.channel().isActive()) {
                ctx.writeAndFlush(dataWithNewline);
                sentCount++;
                logger.debug("Sent data to flink client: {}", jsonData);
            } else {
                // 移除不活跃的连接
                flinkClients.remove(ctx);
            }
        }

        logger.info("Successfully sent data to {} flink clients", sentCount);
    }

    public static void main(String[] args) throws Exception {
        // 客户端连接端口和 Flink 连接端口
        int clientPort = 8888;  // 客户端连接这个端口发送商品数据
        int flinkPort = 9999;   // Flink 连接这个端口接收商品数据

        logger.info("Starting Product Server...");
        logger.info("Client port: {}", clientPort);
        logger.info("Flink port: {}", flinkPort);

        new ProductServer(clientPort, flinkPort).run();
    }
}